#include "Avalon/Threading/Task.hpp"
#include <deque>
#include <iostream>
#include <stack>
#include <boost/enable_shared_from_this.hpp>
#include <boost/exception/diagnostic_information.hpp> 
#include <boost/exception_ptr.hpp> 
#include <boost/scoped_ptr.hpp>
#pragma warning(disable : 4267)
#include <boost/thread.hpp>
#pragma warning(default : 4267)
#include "Avalon/Base/Singleton.hpp"

using namespace Avalon;
using namespace Avalon::Threading;
using namespace boost;
using namespace boost::posix_time;
using namespace std;

#ifdef _DEBUG
namespace {
  void Cleanup(Task* task) {}
  thread_specific_ptr<Task> currentTask(Cleanup);
}
#endif

class Task::TaskThread : public enable_shared_from_this<TaskThread> {
  public:

    TaskThread();

    virtual ~TaskThread();

    bool SetTask(Task* task);

    void ResetTask(Task* task);

    void StopTaskProcessing();

  private:
    Task* m_task;
    scoped_ptr<thread> m_thread;
    bool m_available;
    recursive_mutex m_mutex;
    condition_variable_any m_taskCondition;

    void Run();
};

class Task::ThreadPool : private Singleton<Task::ThreadPool> {
  public:

    virtual ~ThreadPool();

    void IncreaseTaskCount();

    void DecreaseTaskCount();

    void AddTask(Task* task);

    void QueueThread(const shared_ptr<TaskThread>& thread);

    void DequeueThread();

    Task* GetNextTask();

    using Singleton<ThreadPool>::GetInstance;

  private:
    friend class Singleton<Task::ThreadPool>;
    stack<shared_ptr<Task::TaskThread> > m_threads;
    deque<Task*> m_tasks;
    int m_threadCount;
    int m_maxThreads;
    mutex m_mutex;

    ThreadPool();

    void RunNextTask();
};

Task::TaskThread::TaskThread()
    : m_task(NULL),
      m_available(true) {}

Task::TaskThread::~TaskThread() {}

bool Task::TaskThread::SetTask(Task* task) {
  lock_guard<recursive_mutex> lock(m_mutex);
  if(!m_available) {
    return false;
  }
  m_task = task;
  m_taskCondition.notify_all();
  if(m_thread == NULL) {
    m_thread.reset(new thread(bind(&TaskThread::Run, shared_from_this())));
  }
  return true;
}

void Task::TaskThread::ResetTask(Task* task) {
  m_task = task;
}

void Task::TaskThread::StopTaskProcessing() {
  {
    lock_guard<recursive_mutex> lock(m_mutex);
    m_available = false;
    m_taskCondition.notify_all();
  }
  m_thread->join();
  m_thread.reset();
}

void Task::TaskThread::Run() {
  static const time_duration LOWER_BOUND_WAIT_TIME = seconds(30);
  static const time_duration UPPER_BOUND_WAIT_TIME = seconds(60);
  ThreadPool& threadPool = ThreadPool::GetInstance();
  unique_lock<recursive_mutex> lock(m_mutex);
  while(true) {
    if(m_task == NULL) {
      time_duration waitTime = LOWER_BOUND_WAIT_TIME + seconds(rand() %
        (UPPER_BOUND_WAIT_TIME - LOWER_BOUND_WAIT_TIME).total_seconds());
      if(!m_available || !m_taskCondition.timed_wait(lock, waitTime) ||
          m_task == NULL) {
        m_available = false;
        threadPool.DequeueThread();
        return;
      }
    }
#ifdef _DEBUG
    currentTask.reset(m_task);
#endif
    m_task->m_mutex.lock();
    while(m_task->m_workCount > 0) {
      m_task->m_mutex.unlock();
      try {
        m_task->ProcessTask();
      } catch(boost::exception&) {
        cerr << "Exception caught:\n";
        cerr << "Diagnostics: " <<
          boost::diagnostic_information(boost::current_exception()) << endl;
      } catch(std::exception& e) {
        cerr << "Exception caught:\n";
        cerr << "Exception Message: " << e.what() << "\n" << endl;
      }
      m_task->m_mutex.lock();
      --m_task->m_workCount;
    }
    m_task->m_queuedCondition.notify_all();
    if(m_task->m_deleteWhenFinished) {
      m_task->m_mutex.unlock();
      delete m_task;
    } else {
      m_task->m_mutex.unlock();
    }
    m_task = NULL;
#ifdef _DEBUG
    currentTask.reset();
#endif
    threadPool.QueueThread(shared_from_this());
  }
}

Task::ThreadPool::ThreadPool()
    : m_threadCount(0) {
  m_maxThreads = static_cast<int>(thread::hardware_concurrency());
}

Task::ThreadPool::~ThreadPool() {
  while(!m_threads.empty()) {
    shared_ptr<TaskThread> thread = m_threads.top();
    m_threads.pop();
    thread->StopTaskProcessing();
  }
}

void Task::ThreadPool::IncreaseTaskCount() {
  lock_guard<mutex> lock(m_mutex);
  ++m_maxThreads;
  if(!m_tasks.empty()) {
    RunNextTask();
  }
}

void Task::ThreadPool::DecreaseTaskCount() {
  lock_guard<mutex> lock(m_mutex);
  --m_maxThreads;
}

void Task::ThreadPool::AddTask(Task* task) {
  lock_guard<mutex> lock(m_mutex);
  m_tasks.push_back(task);
  RunNextTask();
}

void Task::ThreadPool::QueueThread(const shared_ptr<TaskThread>& thread) {
  lock_guard<mutex> lock(m_mutex);
  if(!m_tasks.empty()) {
    Task* task = m_tasks.front();
    m_tasks.pop_front();
    thread->ResetTask(task);
  } else {
    m_threads.push(thread);
  }
}

void Task::ThreadPool::DequeueThread() {
  lock_guard<mutex> lock(m_mutex);
  --m_threadCount;
}

void Task::ThreadPool::RunNextTask() {
  while(!m_tasks.empty() && !m_threads.empty()) {
    shared_ptr<TaskThread> thread = m_threads.top();
    m_threads.pop();
    Task* frontTask = m_tasks.front();
    if(thread->SetTask(frontTask)) {
      m_tasks.pop_front();
    }
  }
  while(!m_tasks.empty() && m_threadCount < m_maxThreads) {
    Task* frontTask = m_tasks.front();
    shared_ptr<TaskThread> thread(new TaskThread());
    if(thread->SetTask(frontTask)) {
      m_tasks.pop_front();
      ++m_threadCount;
    }
  }
}

Task::Task(bool deleteWhenFinished)
    : m_deleteWhenFinished(deleteWhenFinished),
      m_workCount(0) {}

Task::~Task() {}

void Task::Enqueue() {
  lock_guard<mutex> lock(m_mutex);
  ++m_workCount;
  if(m_workCount == 1) {
    ThreadPool::GetInstance().AddTask(this);
  }
}

void Task::Wait() {
#ifdef _DEBUG
  assert(currentTask.get() != this);
#endif
  unique_lock<mutex> lock(m_mutex);
  while(m_workCount > 0) {
    m_queuedCondition.wait(lock);
  }
}

void Task::IncreaseTaskCount() {
  ThreadPool::GetInstance().IncreaseTaskCount();
}

void Task::DecreaseTaskCount() {
  ThreadPool::GetInstance().DecreaseTaskCount();
}
